package com.nekonomics.lakehouse.dim
import com.nekonomics.lakehouse.common.FlinkRunner._
import com.nekonomics.lakehouse.common.util.SqlUtil._


object ProductSkuDim {
  def main(args: Array[String]): Unit = {
    implicit val config: SqlRunnerConfig = SqlRunnerConfig(groupId = "sku_dim", port = 8081, ckp=true)

    runSql { (_, tableEnv) =>
      tableEnv.executeSql(
          sql"""
               |CREATE TABLE IF NOT EXISTS my_catalog.dim.product_sku_dim  (
               |  sku_id BIGINT PRIMARY KEY NOT ENFORCED,
               |  spu_id BIGINT,
               |  sku_price DECIMAL(10,2),
               |  is_active BOOLEAN,
               |  product_name STRING,
               |  merchant_id BIGINT,
               |  category_id BIGINT,
               |  is_deleted BOOLEAN
               |);
               |""".stripMargin
      )

      tableEnv.executeSql(
        sql"""
             |INSERT INTO dim.product_sku_dim
             |SELECT
             |  sku.id AS sku_id,
             |  sku.spu_id,
             |  sku.price AS sku_price,
             |  sku.is_active,
             |  spu.name AS product_name,
             |  spu.merchant_id,
             |  spu.category_id,
             |  sku.is_deleted
             |FROM ods.sku_ods sku
             |JOIN ods.spu_ods spu ON sku.spu_id = spu.id
             |-- WHERE sku.is_deleted = false AND spu.is_deleted = false;
             |""".stripMargin
      )

    }
  }
}
